package site.teamo.biu.flink;

import cn.hutool.core.io.resource.ResourceUtil;
import com.alibaba.fastjson2.JSON;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import site.teamo.biu.flink.sdk.job.JobContext;
import site.teamo.biu.flink.sdk.job.JobInfo;

public class JaninoRuntimeCompiler {
    public static void main(String[] args) throws Exception {
        // 要编译的Java源代码
        JobInfo jobInfo = JSON.parseObject(ResourceUtil.readUtf8Str("job.json"), JobInfo.class);
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        new JaninoRuntimeCompiler().fillJob(environment, new JobContext(jobInfo));
    }

    public void fillJob(org.apache.flink.streaming.api.environment.StreamExecutionEnvironment environment,site.teamo.biu.flink.sdk.job.JobContext context){
        //        org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<java.lang.Integer> source =environment.addSource(new site.teamo.biu.flink.function.source.RandomNumberSource("source", context)).name("source").setParallelism(1);
//        org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<java.lang.Object> filter = source.process(new site.teamo.biu.flink.function.process.FilterFunction("filter", context)).name("filter").setParallelism(1).returns(Object.class);
//        org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<java.lang.Object> map = filter.process(new site.teamo.biu.flink.function.process.ScriptFunction("map", context)).name("map").setParallelism(1).returns(Object.class);
//        map.addSink(new site.teamo.biu.flink.function.sink.ConsolePrintSink("sink", context)).name("sink").setParallelism(1);
    }
}
